package cn.doitedu.demo2;


import cn.doitedu.beans.UserAction;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hive.ql.udf.UDFSecond;

import java.io.IOException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * 规则ID：rule-003
 * 发生X行为3次以上
 * 发生 Y行为 且 properties['url'] 符合一个正则表达式:  '/a[\\d+]y.html'，就给他发消息
 */
public class RuleCalculator_03 implements RuleCalculator {

    ValueState<Integer> state;
    String ruleId = "rule-003";

    JSONObject message;

    @Override
    public void init(RuntimeContext runtimeContext) throws IOException {
        state = runtimeContext.getState(new ValueStateDescriptor<Integer>("rule-003-cnt", Integer.class));

        message = new JSONObject();
        message.put("rule_id",ruleId);


    }

    @Override
    public void calc(UserAction userAction, Collector<String> out) throws IOException {

        String eventId = userAction.getEvent_id();

        if (eventId.equals("X")) {
            state.update(state.value() == null ? 1 : state.value() + 1);
        }


        if (eventId.equals("Y")) {
            String url = userAction.getProperties().get("url");
            Pattern pattern = Pattern.compile("/a\\[\\d+]y.html");
            Matcher matcher = pattern.matcher(url);
            boolean b = matcher.matches();

            if(b  && state.value() != null && state.value()>3){
                message.put("user_id",userAction.getUser_id());
                message.put("hit_time",userAction.getAction_time());

                out.collect(message.toJSONString());
            }

        }


    }

    @Override
    public String ruleId() {
        return ruleId;
    }


    public static void main(String[] args) {
        Pattern pattern = Pattern.compile("/a\\d+y.html");
        Matcher matcher = pattern.matcher("/a123598y.html");
        System.out.println(matcher.matches());


    }


}
